package com.yanduo.report

import com.yanduo.beans.Log
import com.yanduo.report.AppAnalyseRpt.{broadcast, outputPath}
import com.yanduo.utils.{JedisPools, RptUtils}
import org.apache.commons.lang3.StringUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession

/**
  *媒体报表改造-字典存入到redis中
  * 优化版本
  * @author Gerry chan
  * 2020/5/4 15:38
  * @version 1.0
  */
object AppDict2RedisV2 {
  def main(args: Array[String]): Unit = {
    if (args.length != 2) {
      println(
        """
          |com.yanduo.report.AppDict2Redis
          |参数：
          |  输入路径
          |  输出路径
        """.stripMargin)
      sys.exit()
    }
    val Array(inputPath, outputPath) = args

    val sparkConf = new SparkConf()
    sparkConf.setAppName(s"${this.getClass.getSimpleName}")
    sparkConf.setMaster("local[*]")

    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(sparkConf)
    val spark = new SparkSession(sc)

    sc.textFile(inputPath)
      .map(_.split(",", -1))
      .filter(_.length>=85).map(Log(_))
      .filter(log => !log.appid.isEmpty || !log.appname.isEmpty)   // appid 与appname 都是空的过滤掉
      .mapPartitions(itr => {
      val jedis = JedisPools.getJedis
      //用map时 返回迭代器，但是由于最后要挂壁redis 连接，所以采用forearch
//      itr.map(log => {
//        var newAppName = log.appname
//        if(StringUtils.isNotEmpty(newAppName)) {
//          // 当日志中appName 为空的时候，就去redis中找，
//          // 如果redis也没有则 给默认值 "未知"
//         val newAppName = jedis.get(log.appid)
//        }
      val parResult = new collection.mutable.ListBuffer[(String, List[Double])]()
      itr.foreach(log => {
        var newAppName = log.appname
        if(StringUtils.isEmpty(newAppName)) {
          newAppName= jedis.get(log.appid)
        }
        val reqList: List[Double] = RptUtils.caculateReq(log.requestmode, log.requestmode)
        val rtbList: List[Double] = RptUtils.caulateRtb(log.iseffective, log.isbilling, log.isbid, log.adorderid, log.iswin,
          log.winprice, log.adpayment)
        //广告：点击与展示
        val showClickList: List[Double] = RptUtils.caculateShowClick(log.requestmode, log.iseffective)
        parResult += ((newAppName, reqList++rtbList++showClickList))
      })
      jedis.close()
      // mapPartitions 需要返回迭代器
      parResult.toIterator
    }).reduceByKey((list1,list2) => {
      list1.zip(list2).map(t=>t._1+t._2)
    }).map(t => t._1+","+t._2.mkString(",")).saveAsTextFile(outputPath)

    sc.stop()
    spark.stop()
  }

}
